package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo18Acc {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setAppName("acc")

    conf.setMaster("local")

    val sc = new SparkContext(conf)

    val studentsRDD: RDD[String] = sc.textFile("data/students.txt")

    var count = 0

    val mapRDD: RDD[String] = studentsRDD.map((stu: String) => {

      //在算子内修改算子外的一个普通变量不会生效
      count += 1
      println(count)

      stu
    })

    mapRDD.foreach(println)

    println(s"count:$count")

    /**
     * 累加器：可以在Executor端对Driver端的累加器进行累加
     * 1、先子啊每一个Executor中局部累加
     * 2、当job执行完成之后 将所有Executor的结果拉取到Driver端合并累加结果
     * 优点：不需要单独启动一个job进行累加统计
     *
     */

    //1、在Driver端定义一个累加器
    val countAcc: LongAccumulator = sc.longAccumulator

    studentsRDD
      .map((stu: String) => {

        //2、在Executor端对累加器进行累加
        countAcc.add(1)

        stu
      })
      .foreach(println)

    //3、在Driver端获取累加结果

    println(s"countAcc:${countAcc.value}")

    //count需要好单独启动一个job进行统计，效率低
    val num: Long = studentsRDD.count()


  }

}
